package com.nl.loginfaildetect;

import com.nl.bean.input.LoginLogEvent;
import com.nl.bean.output.Warning;
import java.util.List;
import java.util.Map;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author shihb
 * @date 2019/12/26 10:26
 * 2秒内登录失败次数超过2次的CEP实现
 * CEP 复杂事件处理
 */
public class LoginFailWithCep {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.setParallelism(1);
    DataStreamSource<String> source = env.readTextFile(
        LoginFailDetect.class.getClassLoader().getResource("LoginLog.csv").getPath());
    //1.读取事件数据，创建简单事件流
    KeyedStream<LoginLogEvent,Long> eventStream = source
        .map(s -> {
          String[] arr = s.split(",");
          long userId = Long.parseLong(arr[0].trim());
          long loginTime = Long.parseLong(arr[3].trim());
          return new LoginLogEvent(userId, arr[1].trim(), arr[2].trim(), loginTime);
        })
       .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginLogEvent>(
            Time.seconds(5)) {
          @Override
          public long extractTimestamp(LoginLogEvent element) {
            return element.getLoginTime() * 1000;
          }
        })
        .keyBy(loginLogEvent->loginLogEvent.getUserId());

    //2.定义CEP匹配模式(Pattern API)
    Pattern<LoginLogEvent, LoginLogEvent> loginFailPattern = Pattern
        .<LoginLogEvent>begin("start").where(new SimpleCondition<LoginLogEvent>(){
          @Override
          public boolean filter(LoginLogEvent loginLogEvent) throws Exception {
            return "fail".equals(loginLogEvent.getLoginState().toLowerCase());
          }
        })
        .next("next").where(new SimpleCondition<LoginLogEvent>() {
          @Override
          public boolean filter(LoginLogEvent loginLogEvent) throws Exception {
            return "fail".equals(loginLogEvent.getLoginState().toLowerCase());
          }
        }).within(Time.seconds(2));


    //3.在事件流上应用模式，得到pattern stream
    PatternStream<LoginLogEvent> patternStream= CEP.pattern(eventStream, loginFailPattern);

    //4.从pattern stream应用select function,检出匹配事件序列
    SingleOutputStreamOperator<Warning> warningStream = patternStream
        .select(new PatternSelectFunction<LoginLogEvent, Warning>() {
          @Override
          public Warning select(Map<String, List<LoginLogEvent>> map) throws Exception {
            LoginLogEvent firstFail = map.get("start").iterator().next();
            LoginLogEvent secondFail = map.get("next").iterator().next();
            return new Warning(firstFail.getUserId(),firstFail.getLoginTime().toString(),secondFail.getLoginTime().toString(),"login fail.");
          }
        });

    warningStream.print();
    env.execute("login fail with CEP job");







  }

}
